package cn.zwx.canal.tools.client;

import cn.hutool.core.util.StrUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

/**
 * @description: canalClient客户端
 * @projectName:learn
 * @see:cn.zwx.canal.tools.config
 * @author:zhangwenxue
 * @createTime:2020/12/19 10:21
 * @version:1.0
 */
@Component
public class CanalClient implements DisposableBean {

    private Logger logger = LoggerFactory.getLogger(CanalClient.class);

    private CanalConnector canalConnector;

    @Value("${canal.host}")
    private String canalHost;
    @Value("${canal.port}")
    private String canalPort;
    @Value("${canal.destination}")
    private String canalDestination;
    @Value("${canal.username:null}")
    private String canalUsername;
    @Value("${canal.password:null}")
    private String canalPassword;
    @Value("${canal.subscribe}")
    private String subscribe;
    @Value("${canal.server.zkServer}")
    private String zkServer;

    @Bean
    public CanalConnector getCanalConnector(){
        // 1.构建对象
//        canalConnector =  CanalConnectors.newClusterConnector(Arrays.asList(new InetSocketAddress(canalHost,Integer.valueOf(canalPort))),
//                canalDestination,canalUsername,canalPassword);
        canalConnector =  CanalConnectors.newClusterConnector(zkServer,
                canalDestination,canalUsername,canalPassword);
        try {
            // 2.创建连接
            canalConnector.connect();
            // 3.订阅客户端 客户端指定会覆盖服务指定配置
            if (StrUtil.isNotBlank(subscribe)){
                canalConnector.subscribe(subscribe);
            }
            // 4.回滚寻找上次中断的位置
            canalConnector.rollback();
            logger.info("canalClient客户端 init success....");
        } catch (Throwable t) {
            logger.error("failed to connect to canal server", t);
            canalConnector.disconnect();
            throw t;
        }

        logger.info("canal客户端启动成功!!!!");
        return canalConnector;
    }

    /**
     * @description 重写销毁方法
     **/
    @Override
    public void destroy() throws Exception {
        if (canalConnector != null) {
            canalConnector.disconnect();
        }
    }
}
